package com.zhang.spark_1.spark_core.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/5 18:51 
 */
object Spark22_RDD_Req {

  def main(args: Array[String]): Unit = {
    //获取spark的连接
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc: SparkContext = new SparkContext(conf)
    /*
    agent.log：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
    需求描述:统计出每一个省份每个广告被点击数量排行的Top3
     */
    val rdd: RDD[String] = sc.textFile("datas/agent.log")
     //1.(时间戳，省份，城市，用户，广告)=>（(省份，广告），1）
     val mapRDD: RDD[((String, String), Int)] = rdd.map(
       line => {
         val datas: Array[String] = line.split(" ")
         ((datas(1), datas(4)), 1)
       }
     )
    //2.（(省份，广告），1）=>（(省份，广告），sum）
    val reduceRDD: RDD[((String, String), Int)] = mapRDD.reduceByKey(_ + _)

    //3.转换结构（（(省份，广告），sum））=>（省份，(广告，sum)）
    val newMapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      case ((prov, ad), sum) => (prov, (ad, sum))
    }
    //4.（省份，(广告，sum)）=>(省份，【(广告，sum)】）
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey()

    //5.组内排序，取前三
    groupRDD.mapValues(
      iter=>{
        iter.toList.sortWith(_._2>_._2).take(3)
      }
    ).collect().foreach(println)
 

    sc.stop()

  }
}
